﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace Flower.Actors
{
    public sealed class Actor<TProcessor, TMessage>
        : IActor<TMessage>
        where TMessage : class, IMessage<TProcessor>
    {
        private static readonly TimeSpan MessageWaitPollInterval = TimeSpan.FromMilliseconds(100);
        private static readonly object TimeoutSignal = new object();

        private readonly IProducerConsumerCollection<TMessage> _queue; 
        private readonly ManualResetEventSlim _canAdd = new ManualResetEventSlim(true);
        private readonly ManualResetEventSlim _canTake = new ManualResetEventSlim(false);
        private readonly int _queueCapacity;
        private bool _isAddingCompleted;

        private readonly Thread _thread;
        private readonly Thread _watchThread;
        private readonly TimeSpan? _processingTimeout;
        private readonly AutoResetEvent _processingStarted = new AutoResetEvent(false);
        private readonly ManualResetEvent _processingFinished = new ManualResetEvent(false);
        private readonly ManualResetEvent _stopWatching = new ManualResetEvent(false);
        private readonly ManualResetEvent _resume = new ManualResetEvent(false);

        private volatile bool _isSuspended;
        private volatile bool _isDisposed;

        public event EventHandler<ActorExceptionEventArgs<TMessage>> Error;

        public IProducerConsumerCollection<TMessage> Queue { get { return _queue; } }

        public Actor(TProcessor processor, IProducerConsumerCollection<TMessage> queue, int queueCapacity = 0, TimeSpan? processingTimeout = null)
        {
            _queue = queue;
            _queueCapacity = queueCapacity;

            _thread = new Thread(Processing) 
            {
                IsBackground = true
            };
            _thread.Start(processor);
            _processingTimeout = processingTimeout;

            if (processingTimeout.HasValue)
            {
                _watchThread = new Thread(Timeout) 
                {
                    IsBackground = true
                };
                _watchThread.Start();
            }
            else
            {
                _watchThread = null;
            }
        }

        public Actor(TProcessor processor, TimeSpan? processingTimeout = null)
            : this(processor, new ConcurrentQueue<TMessage>(), 0, processingTimeout)
        {
        }

        private void Timeout()
        {
            var events = new WaitHandle[]
            {
                _processingStarted,
                _stopWatching
            };

            while (true)
            {
                int signaled = WaitHandle.WaitAny(events);

                switch (signaled)
                {
                    case 0:
                        if (!_processingFinished.WaitOne(_processingTimeout.Value))
                        {
                            _thread.Abort(TimeoutSignal);
                        }
                        break;
                    case 1:
                        return;
                }
            }
        }

        private void Processing(object processorObj)
        {
            var processor = (TProcessor)processorObj;

            TMessage msg = null;

            while (true)
            {
                if (msg == null)
                {
                    _canTake.Wait();

                    lock (_queue)
                    {
                        if (_isAddingCompleted)
                        {
                            break;
                        }

                        if (!_queue.TryTake(out msg))
                        {
                            _canTake.Reset();
                            continue;
                        }

                        _canAdd.Set();
                    }
                }

                try
                {
                    _processingFinished.Reset();
                    _processingStarted.Set();
                    msg.Process(processor);
                    _processingFinished.Set();
                    
                    msg = null;
                }
                catch (ActorSuspendException)
                {
                    _processingFinished.Set();
                    _isSuspended = true;
                    _resume.WaitOne();
                    _isSuspended = false;
                    _resume.Reset();
                }
                catch (Exception ex)
                {
                    msg = null;

                    try
                    {
                        var eh = Error;
                        if (eh != null)
                        {
                            try
                            {
                                eh.Invoke(processor, new ActorExceptionEventArgs<TMessage>(msg, ex));
                            }
                            catch (Exception) { }
                        }

                        var threadAbort = ex as ThreadAbortException;

                        if (threadAbort != null && threadAbort.ExceptionState == TimeoutSignal)
                        {
                            Thread.ResetAbort();
                        }
                        else
                        {
                            _processingFinished.Set();
                        }
                    }
                    catch (ThreadAbortException threadAbort)
                    {
                        if (threadAbort.ExceptionState == TimeoutSignal)
                        {
                            Thread.ResetAbort();
                        }
                    }
                }

                lock (_queue)
                {
                    if (_isAddingCompleted)
                    {
                        break;
                    }
                }
            }
        }

        public void Send(TMessage message)
        {
            if (_isDisposed) return;

            _canAdd.Wait();

            lock (_queue)
            {
                _queue.TryAdd(message);

                _canTake.Set();
                
                if (_queueCapacity > 0 && _queue.Count == _queueCapacity)
                {
                    _canAdd.Reset();
                }
            }
        }

        public void Dispose()
        {
            if (!_isDisposed)
            {
                _isDisposed = true;

                if (!_isAddingCompleted)
                {
                    Stop();
                }
            }
        }

        public void Stop()
        {
            try
            {
                lock (_queue)
                {
                    _isAddingCompleted = true;
                    _canTake.Set();
                }

                _resume.Set();
                _thread.Join();

                if (_watchThread != null)
                {
                    _stopWatching.Set();
                    _watchThread.Join();
                }
            }
            catch (Exception) { }
        }

        public void Resume()
        {
            _resume.Set();
        }

        public bool WaitAllMessages(TimeSpan? timeout)
        {
            var waitStart = DateTime.Now;

            while (!timeout.HasValue || (DateTime.Now - waitStart) < timeout.Value)
            {
                if (_queue.Count == 0)
                {
                    return true;
                }

                Thread.Sleep(MessageWaitPollInterval);
            }

            return false;
        }

        public bool IsSuspended { get { return _isSuspended; } }
    }
}
